package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"time"

	"github.com/segmentio/kafka-go"
)

func main() {
	// 配置Kafka消费者 - 使用正确的远程地址
	config := kafka.ReaderConfig{
		Brokers:   []string{"10621767cokk0.vicp.fun:58565"},
		Topic:     "mumu",
		GroupID:   "go-kafka-consumer-" + time.Now().Format("20060102150405"),
		MinBytes:  10e3,
		MaxBytes:  10e6,
	}

	// 创建消费者
	reader := kafka.NewReader(config)
	defer reader.Close()

	fmt.Println("正在连接到Kafka服务器...")

	// 测试连接
	conn, err := kafka.Dial("tcp", "10621767cokk0.vicp.fun:58565")
	if err != nil {
		log.Fatalf("无法连接到Kafka服务器: %v", err)
	}
	defer conn.Close()
	fmt.Println("成功连接到Kafka服务器")

	// 设置优雅退出
	sigchan := make(chan os.Signal, 1)
	signal.Notify(sigchan, os.Interrupt)

	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		<-sigchan
		cancel()
	}()

	fmt.Println("开始消费消息...")
	for {
		msg, err := reader.ReadMessage(ctx)
		if err != nil {
			if err == context.Canceled {
				fmt.Println("消费者正在关闭...")
				break
			}
			log.Printf("消费错误: %v\n", err)
			time.Sleep(5 * time.Second) // 添加重试延迟
			continue
		}
		fmt.Printf("收到消息 [主题:%s 分区:%d 偏移量:%d]: %s\n",
			msg.Topic, msg.Partition, msg.Offset, string(msg.Value))
	}

	fmt.Println("消费者已退出")
}